package org.example.insert_into;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonInsertIntoUpdate {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(conf);
        env.setParallelism(1);
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        TableEnvironment tableEnv = StreamTableEnvironment.create(env);
        
        // 添加Paimon所需的配置
        tableEnv.getConfig().getConfiguration().setString("table.exec.sink.upsert-materialize", "NONE");

        // 定义SQL语句
        String createCatalogSql = "CREATE CATALOG paimon WITH (\n" +
                "    'type' = 'paimon',\n" +
                "    'warehouse' = 'file:///tmp/paimon'\n" +
                ");";
        // 打印SQL语句
        System.out.println("执行以下SQL创建Paimon Catalog：\n" + createCatalogSql);
        tableEnv.executeSql(createCatalogSql);

        // 定义SQL语句
        String createBehaviorGenSql = "CREATE TABLE behavior_gen (\n" +
                "    user_id STRING,\n" +
                "    item_id STRING,\n" +
                "    behavior BIGINT,\n" +
                "    dt AS '2025-01-11',\n" +
                "    eventtime TIMESTAMP_LTZ(3)\n" +
                ") WITH (\n" +
                "  'connector' = 'datagen',\n" +
                "  'rows-per-second' = '200'\n" + 
                ");\n";
        // 打印SQL语句
        System.out.println("执行以下SQL创建数据生成表：\n" + createBehaviorGenSql);
        tableEnv.executeSql(createBehaviorGenSql);

        // 定义SQL语句
        String dropTableSql = "DROP TABLE IF EXISTS `paimon`.`default`.`user_behavior_cnt`";
        // 打印SQL语句
        System.out.println("执行以下SQL删除已存在的表：\n" + dropTableSql);
        tableEnv.executeSql(dropTableSql);

        // 定义SQL语句
        String createPaimonTableSql = "-- 创建paimon表\n" +
                "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`user_behavior_cnt` (\n" +
                "    dt STRING,\n" +
                "    behavior BIGINT,\n" +
                "    cnt      BIGINT,\n" +
                "    PRIMARY KEY (dt, behavior) NOT ENFORCED\n" +
                ") PARTITIONED BY (dt)\n" +
                ";";
        // 打印SQL语句
        System.out.println("执行以下SQL创建Paimon表：\n" + createPaimonTableSql);
        tableEnv.executeSql(createPaimonTableSql);

        // 修复SQL，确保GROUP BY与SELECT中计算的列一致
        final String insertStatement = "INSERT INTO `paimon`.`default`.`user_behavior_cnt`\n" +
                "SELECT\n" +
                "    dt\n" +
                "    , abs(behavior % 10) AS behavior\n" +
                "    , count(1) as cnt\n" +
                "FROM \n" +
                "    behavior_gen\n" +
                "GROUP BY\n" +
                "    dt, abs(behavior % 10)\n" +
                ";";
        // 打印SQL语句
        System.out.println("执行以下SQL插入数据：\n" + insertStatement);
        tableEnv.executeSql(insertStatement);
        
        // 添加查询以验证结果
        System.out.println("查询结果表中的数据：");
        String querySql = "SELECT * FROM `paimon`.`default`.`user_behavior_cnt`";
        tableEnv.executeSql(querySql).print();
    }
}
